package com.wang.wx.service.impl;

import com.wang.wx.constants.BatchConstants;
import com.wang.wx.entity.EmailUser;
import com.wang.wx.entity.LyUser;
import com.wang.wx.entity.TwitterUser;
import com.wang.wx.entity.WxUser;
import com.wang.wx.service.EmailUserService;
import com.wang.wx.service.LyUserService;
import com.wang.wx.service.TwitterUserService;
import com.wang.wx.service.WxUserService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Slf4j
public class BatchProcessor {


    public static void processWxQueue(WxUserService wxUserService, int batchSize) {
        BlockingQueue<WxUser> queue = BatchConstants.WX_QUEUE;
        batchSize = batchSize / 10;
        List<WxUser> buffer = new ArrayList<>(batchSize);

        try {
            while (true) {
                WxUser user = queue.poll(200, TimeUnit.MILLISECONDS);
                if (user != null) {
                    buffer.add(user);
                }

                if (buffer.size() >= batchSize || (BatchConstants.WX_SIGNAL_COMPLETION.get() && !buffer.isEmpty() && queue.isEmpty())) {
                    //1、如果buffer队列满了，走这一步
                    //2、如果buffer队列没满，但是主线程已经读完数据了，并且阻塞队列已经空了，并且buffer还有剩余一点数据，走这一步
                    wxUserService.saveBatch(buffer);
                    buffer.clear();
                    if (BatchConstants.WX_SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                        break;
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("批处理线程被中断", e);
        }
    }

    public static void processTwitterQueue(TwitterUserService twitterUserService, int batchSize) {
        BlockingQueue<TwitterUser> queue = BatchConstants.TWITTER_QUEUE;
        batchSize = batchSize / 10;
        List<TwitterUser> buffer = new ArrayList<>(batchSize);

        try {
            while (true) {
                TwitterUser twitterUser = queue.poll(200, TimeUnit.MILLISECONDS);
                if (twitterUser != null) {
                    buffer.add(twitterUser);
                }

                if (buffer.size() >= batchSize || (BatchConstants.TWITTER_SIGNAL_COMPLETION.get() && !buffer.isEmpty() && queue.isEmpty())) {
                    //1、如果buffer队列满了，走这一步
                    //2、如果buffer队列没满，但是主线程已经读完数据了，并且阻塞队列已经空了，并且buffer还有剩余一点数据，走这一步
                    twitterUserService.saveBatch(buffer);
                    buffer.clear();
                    if (BatchConstants.TWITTER_SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                        break;
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("批处理线程被中断", e);
        }
    }

    public static void processLyQueue(LyUserService lyUserService, int batchSize) {
        BlockingQueue<LyUser> queue = BatchConstants.LY_QUEUE;
        batchSize = batchSize / 10;
        List<LyUser> buffer = new ArrayList<>(batchSize);

        try {
            while (true) {
                LyUser lyUser = queue.poll(200, TimeUnit.MILLISECONDS);
                if (lyUser != null) {
                    buffer.add(lyUser);
                }

                if (buffer.size() >= batchSize || (BatchConstants.LY_SIGNAL_COMPLETION.get() && !buffer.isEmpty() && queue.isEmpty())) {
                    //1、如果buffer队列满了，走这一步
                    //2、如果buffer队列没满，但是主线程已经读完数据了，并且阻塞队列已经空了，并且buffer还有剩余一点数据，走这一步
                    try {
                        lyUserService.saveBatch(buffer);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    buffer.clear();
                    if (BatchConstants.LY_SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                        break;
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("批处理线程被中断", e);
        }
    }


    public static void processEmailQueue(EmailUserService emailUserService, int batchSize) {
        BlockingQueue<EmailUser> queue = BatchConstants.EMAIL_QUEUE;
        batchSize = batchSize / 10;
        List<EmailUser> buffer = new ArrayList<>(batchSize);

        try {
            while (true) {
                EmailUser emailUser = queue.poll(200, TimeUnit.MILLISECONDS);
                if (emailUser != null) {
                    buffer.add(emailUser);
                }

                if (buffer.size() >= batchSize || (BatchConstants.EMAIL_SIGNAL_COMPLETION.get() && !buffer.isEmpty() && queue.isEmpty())) {
                    //1、如果buffer队列满了，走这一步
                    //2、如果buffer队列没满，但是主线程已经读完数据了，并且阻塞队列已经空了，并且buffer还有剩余一点数据，走这一步
                    try {
                        emailUserService.saveBatch(buffer);
//                        log.info("已入库:{}", buffer.size());
                    } catch (Exception e) {
                        for (EmailUser user : buffer) {
                            try {
                                emailUserService.save(user);
                            } catch (Exception e1) {
//                                log.error("{}", user);
//                                e1.printStackTrace();
                            }
                        }
                    }
                    buffer.clear();
                    if (BatchConstants.EMAIL_SIGNAL_COMPLETION.get() && queue.isEmpty()) {
                        break;
                    }
                }
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            log.error("批处理线程被中断", e);
        }
    }

}